#include "mysql_session.h"
#include <iostream>
#include <string>
#include <assert.h>
#include <boost/asio.hpp>
#include "globals.h"

using namespace std;
using namespace boost;
using namespace boost::asio;


MysqlSession::MysqlSession(boost::asio::io_service& io_service) :
	TcpSession(io_service), buf_read(0)
{
	handles[EET_REGISTER_URL] = boost::bind(
		&MysqlSession::onRegisterUrl, this, _1);
	handles[EET_REGISTER_USER] = boost::bind(
		&MysqlSession::onRegisterUser, this, _1);
	handles[EET_REQUEST_FRIEND] = boost::bind(
		&MysqlSession::onRequestFriend, this, _1);
}

MysqlSession::pointer MysqlSession::clone()
{
	return pointer(new MysqlSession(socket_.io_service()));
};

void MysqlSession::process_read()
{
	BaseEvent *event = (BaseEvent*) buf;
	HandleMap::iterator itr;
	itr = handles.find((E_EVENT_TYPE)event->type);
	if (itr == handles.end())
	{
		cerr << "Error in event type: " << event->type << endl;
		return;
	}
	itr->second(buf);
};

void MysqlSession::start()
{
	buf_read = 0;
	async_read(socket_, boost::asio::buffer(buf, sizeof(BaseEvent)),
		strand_.wrap(boost::bind(&TcpSession::handle_read,
		shared_from_this(), 
		boost::asio::placeholders::error,
		boost::asio::placeholders::bytes_transferred)));
}

void MysqlSession::handle_read(const boost::system::error_code& err,
		std::size_t bytes_transferred)
{
	if (!err)
	{
		buf_read += bytes_transferred;
		if (buf_read < sizeof(BaseEvent))
		{
			async_read(socket_, boost::asio::buffer(buf + buf_read,
					sizeof(BaseEvent) - buf_read),
				strand_.wrap(boost::bind(&TcpSession::handle_read,
				shared_from_this(), 
				boost::asio::placeholders::error,
				boost::asio::placeholders::bytes_transferred)));
			return;
		}
		BaseEvent *event = (BaseEvent*) buf;
		if (event->size < sizeof(BaseEvent) || event->size > MAX_EVENT_LENGTH)
		{
			std::cerr << "Error in event size: " << event->size << std::endl;
			buf_read = 0;
			async_read(socket_, boost::asio::buffer(buf, sizeof(BaseEvent)),
				strand_.wrap(boost::bind(&TcpSession::handle_read,
				shared_from_this(), 
				boost::asio::placeholders::error,
				boost::asio::placeholders::bytes_transferred)));
			return;
		}
		if (buf_read < event->size)
		{
			async_read(socket_, boost::asio::buffer(buf + buf_read,
					event->size - buf_read),
				strand_.wrap(boost::bind(&TcpSession::handle_read,
				shared_from_this(), 
				boost::asio::placeholders::error,
				boost::asio::placeholders::bytes_transferred)));
			cerr << "buf_read < event->size" << endl;
			return;
		}
		assert(event->size == buf_read);
		if (event->size == buf_read)
		{
			process_read();
			buf_read = 0;
			async_read(socket_, boost::asio::buffer(buf, sizeof(BaseEvent)),
				strand_.wrap(boost::bind(&TcpSession::handle_read,
				shared_from_this(), 
				boost::asio::placeholders::error,
				boost::asio::placeholders::bytes_transferred)));
			return;
		}
	}
	else
	{
		std::cerr << "Error: " << err.message() << "\n";
		std::cerr << "MysqlSession exit..." << std::endl;
	}
}

void MysqlSession::onRegisterUrl(void *buffer) {
	cerr << "MysqlSession::onRegisterUrl" << endl;
	RequestRegisterUrl* msg = (RequestRegisterUrl*) buffer;
	stringstream ss;
	ss << "insert into url_cache (url_str, client_id) values "
		<< "( \"" << msg->url << "\", "
		<< user_guid << ") on duplicate key update guid=guid; ";

	cerr << ss.str() << endl;

	if (mysql_query(mysql, ss.str().c_str()) != 0) {
		cerr << "mysql error: " << mysql_error(mysql) << endl;
	}
	
	RespondRegisterUrl resp;
	resp.state = 0;
	async_write(string((char*)&resp, resp.size));
}

void MysqlSession::onRegisterUser(void *buffer) {
	RequestRegisterUser *msg = (RequestRegisterUser*) buffer;
	stringstream ss;
	ss << (int)msg->local_ip[3] << "." << (int)msg->local_ip[2]
			<< "." << (int)msg->local_ip[1] << "." << (int)msg->local_ip[0];
	string local_ip = ss.str();
	cout << "get local ip: " << local_ip << endl;
	tcp::endpoint endpoint = socket_.remote_endpoint();	
	ip::address address = endpoint.address();
	string global_ip = address.to_v4().to_string();
	cout << "get global ip: " << global_ip << endl;
	cout << "get guid: " << msg->guid << endl;

	user_guid = msg->guid;
	if (user_guid == 0)
	{
		ss.str("");
		ss << "insert into client_cache "
			<< "(global_ip_addr, local_ip_addr, port, cache_date) values "
			<< "(\"" << global_ip << "\", "
			<< "\"" << local_ip << "\", "
			<< msg->port << ", " << "now() ); ";
		if ( (mysql_query(mysql, ss.str().c_str())) != 0) {
			cerr << "mysql_error:" << mysql_error(mysql) << endl;
			return;
		}
		if ( (mysql_query(mysql, "SELECT last_insert_id(); ")) != 0) {
			cerr << "mysql_error:" << mysql_error(mysql) << endl;
			return;
		}
		MYSQL_RES * result = mysql_store_result(mysql);
		MYSQL_ROW row = mysql_fetch_row(result);
		user_guid = atol(row[0]);
		RespondRegisterUser resp;
		resp.state = 0;
		resp.guid = user_guid;
		cout << "new guid: " << user_guid << endl;
		async_write(string((char*)&resp, resp.size));
		mysql_free_result(result);
	} else {
		ss.str("");
		ss << "update client_cache set "
			<< "global_ip_addr = \"" << global_ip << "\", "
			<< "local_ip_addr = \"" << local_ip << "\", "
			<< "port = " << msg->port << ", "
			<< "cache_date = now() "
			<< "where guid = " << user_guid << "; ";
		if ( (mysql_query(mysql, ss.str().c_str())) != 0) {
			cerr << "msqyl_error:" << mysql_error(mysql) << endl;
			return;
		}
		RespondRegisterUser resp;
		resp.state = 0;
		resp.guid = user_guid;
		async_write(string((char*)&resp, resp.size));
	}
}

void MysqlSession::onRequestFriend(void *buffer)
{
	cout << "MysqlSession::onRequestFriend" << endl;
	RequestFriend* msg = (RequestFriend*) buffer;
	stringstream ss;
	ss << "select global_ip_addr, local_ip_addr, port from client_cache, "
		<< "url_cache where client_cache.guid = url_cache.client_id "
		<< "and url_str = \"" << msg->url << "\" "
		<< "and client_cache.guid != " << user_guid << " "
		<< "order by cache_date desc limit 20; ";
	cerr << ss.str() << endl;
	if (mysql_query(mysql, ss.str().c_str()) != 0) {
		cerr << "mysql error: " << mysql_error(mysql) << endl;
		return;
	}

	MYSQL_RES* result = mysql_store_result(mysql);
	if (result == NULL) {
		cerr << "result == 0" << endl;
		return;
	}
	MYSQL_ROW row;
	long row_count = mysql_num_rows(result);
	RespondFriend resp;
	resp.peer_count = 0;
	while (resp.peer_count < row_count)
	{
		row = mysql_fetch_row(result);
		if (row == NULL) {
			cerr << "row == 0" << endl;
			return;
		}
		int global_ip[4];
		int local_ip[4];
		int port;
		sscanf(row[0], "%d . %d . %d . %d", &global_ip[3], &global_ip[2],
			&global_ip[1], &global_ip[0]);
		sscanf(row[1], "%d . %d . %d . %d", &local_ip[3], &local_ip[2],
			&local_ip[1], &local_ip[0]);
		sscanf(row[2], "%d", &port);

		cerr << "global ip: " << global_ip[3] << "." << global_ip[2] << "."
			<< global_ip[1] << "." << global_ip[0] << endl;
		cerr << "local ip: " << local_ip[3] << "." << local_ip[2] << "."
			<< local_ip[1] << "." << local_ip[0] << endl;
		cerr << port << endl;
	
		resp.peers[resp.peer_count].global_ip[3] = (uint8_t)global_ip[3];
		resp.peers[resp.peer_count].global_ip[2] = (uint8_t)global_ip[2];
		resp.peers[resp.peer_count].global_ip[1] = (uint8_t)global_ip[1];
		resp.peers[resp.peer_count].global_ip[0] = (uint8_t)global_ip[0];

		resp.peers[resp.peer_count].local_ip[3] = (uint8_t)local_ip[3];
		resp.peers[resp.peer_count].local_ip[2] = (uint8_t)local_ip[2];
		resp.peers[resp.peer_count].local_ip[1] = (uint8_t)local_ip[1];
		resp.peers[resp.peer_count].local_ip[0] = (uint8_t)local_ip[0];
		resp.peers[resp.peer_count].port = (uint16_t) port;
		resp.peer_count ++;
	}
	async_write(string((char*)&resp, resp.size));
}
